package com.atguigu.api2

import com.atguigu.api.SensorReading
import org.apache.flink.api.java.tuple.{Tuple, Tuple1}
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector


/**
 * @description: 创建不同类型的窗口,窗口必须跟着后面函数才算是完整的一次计算
 * @time: 2020/6/21 13:10
 * @author: baojinlong
 **/
object WindowApiTest2 {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    // 设置周期生成watermark时间间隔,默认200毫秒
    environment.getConfig.setAutoWatermarkInterval(100L)
    val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data")

    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })
      // .assignAscendingTimestamps(_.timestamp * 1000L)  // 升序时间设置
      //      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) { // 最乱序时间设置
      //        override def extractTimestamp(t: SensorReading): Long = {
      //          t.timestamp * 1000L
      //        }
      //      })
      // 自定义写法
      .assignTimestampsAndWatermarks(new MyWMAssigner(1000L)) // 可以继续设置并行度

    val resultDataStream: DataStream[(Long, Int)] = dataStream
      .keyBy("id") // window包括TimeWork和CountWindow
      //.window(EventTimeSessionWindows.withGap(Time.minutes(1))) // 会话窗口
      //.timeWindow(Time.hours(1), Time.minutes(1)) // 传一个参数为滚动窗口,两个为滑动窗口
      //.window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(2))) // 滚动窗口
      //.countWindow(10, 2) // 计数窗口
      //.timeWindow(Time.milliseconds(1000))
      // 注意如果时间没达到则没有数据输出
      //.timeWindow(Time.seconds(15))
      .timeWindow(Time.seconds(15), Time.seconds(5)) // 滑动窗口
      // 对迟到数据进行修正.在桶计算完成后暂时不关闭,先输出一个不精确的结果数据
      .allowedLateness(Time.minutes(1))
      // 把没有纳入的数据放入别的流,在延时一分钟后的数据直接作为侧输出流
      .sideOutputLateData(new OutputTag[SensorReading]("late-data"))
      // .reduce(new MyReduce)
      .apply(new MyWindowFunction) //全窗口案例

    // 取出溢出数据 三个地方设置迟到数据 assignTimestampsAndWatermarks  allowedLateness  sideOutputLateData getSideOutput
    val value: DataStream[SensorReading] = resultDataStream.getSideOutput(new OutputTag[SensorReading]("late-data"))
    // 打印结果
    dataStream.print("dataStream")
    resultDataStream.print("resultDataStream")
    environment.execute("sink simple test job")
  }
}

// 自定义一个全窗口函数
class MyWindowFunction extends WindowFunction[SensorReading, (Long, Int), Tuple, TimeWindow] {
  override def apply(key: Tuple, window: TimeWindow, input: Iterable[SensorReading], out: Collector[(Long, Int)]): Unit = {
    out.collect((window.getStart, input.size))
  }
}

// 带key的转换
class MyWindowFunction33 extends WindowFunction[SensorReading, (String, Long, Int), Tuple, TimeWindow] {
  override def apply(key: Tuple, window: TimeWindow, input: Iterable[SensorReading], out: Collector[(String, Long, Int)]): Unit = {
    val id: String = key.asInstanceOf[Tuple1[String]].f0
    out.collect((id, window.getStart, input.size))
  }
}


// 自定义一个周期生成watermark的Assigner,默认是200毫秒
  class MyWMAssigner(lateness: Long) extends AssignerWithPeriodicWatermarks[SensorReading] {
  // 需要两个关键参数,延时时间和当前所有数据中最大的时间戳
  var maxTs: Long = Long.MinValue + lateness

  override def getCurrentWatermark: Watermark = new Watermark(maxTs - lateness)

  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    maxTs = maxTs.max(t.timestamp * 1000L)
    t.timestamp * 1000L

  }
}

// 自定义一个断点式成watermark的Assigner
class MyWMAssigner2 extends AssignerWithPunctuatedWatermarks[SensorReading] {
  // 需要两个关键参数,延时时间和当前所有数据中最大的时间戳
  val lateness: Long = 1000L

  override def checkAndGetNextWatermark(t: SensorReading, l: Long): Watermark = {
    if ("sensor_1".equals(t.id)) {
      new Watermark(l - lateness)
    } else {
      null
    }
  }

  override def extractTimestamp(t: SensorReading, l: Long): Long = t.timestamp * 1000L
}